<?php
/**
 *  * *    author 韩天尊
 * *    Date: 2022/4/12
 * *    Email: <724700483@qq.com>
 */

namespace App\Process;
use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Kafka\Config\ConsumerConfig;
use EasySwoole\Kafka\Exception\Config;
use EasySwoole\Kafka\Exception\Exception;
use EasySwoole\Kafka\Kafka;

class KafkaConsumer extends AbstractProcess
{
    protected function run($arg)
    {

        $instance = \EasySwoole\EasySwoole\Config::getInstance();//配置获取
        $metadataBrokerList = $instance->getConf("KAFKA.url");
        $version = $instance->getConf("KAFKA.version");
        $process = $instance->getConf("KAFKA.process");
        foreach ($process as $item){
            $item = arrayTransitionObject($item);
            go(function () use($metadataBrokerList, $version, $item){
                $config = new ConsumerConfig();
                $config->setRefreshIntervalMs($item->refreshIntervalMs);//轮询(毫秒 1000=1秒)
                $config->setMetadataBrokerList($metadataBrokerList);
                $config->setBrokerVersion($version);
                $config->setGroupId($item->groupId);//标识消费者是谁 不一致将会导致会把所有历史任务又全处理一遍

                $config->setTopics(objectTransitionArray($item->topics));
                $config->setOffsetReset($item->offsetReset);//latest 最新   earliest 最早

                $kafka = new Kafka($config);
                // 设置消费回调
                $kafka->consumer()->subscribe( function ($topic, $partition, $message) {
                    $message = arrayTransitionObject($message['message']);
                    //var_dump($message->key);
                    //var_dump($message->value);
                    $jobObj = json_decode($message->value);
                    call_user_func_array(array(new $jobObj->class, $jobObj->action), objectTransitionArray($jobObj->data));
                });
            });
        }

        echo($this->getProcessName()." runing... ".PHP_EOL);
    }
}